package com.kqkj.dyrule.engine.router;

import com.kqkj.dyrule.engine.beans.EventBean;
import com.kqkj.dyrule.engine.beans.EventParam;
import com.kqkj.dyrule.engine.beans.EventSequenceParam;
import com.kqkj.dyrule.engine.beans.RuleConditions;
import com.kqkj.dyrule.engine.queryService.ClickHouseQueryServiceImpl;
import com.kqkj.dyrule.engine.queryService.HbaseQueryServiceImpl;
import com.kqkj.dyrule.engine.queryService.StateQueryServiceImpl;
import com.kqkj.dyrule.engine.util.ConnectionUtils;
import com.kqkj.dyrule.engine.util.EventParamComparator;
import com.kqkj.dyrule.engine.util.SegmentQueryUtil;
import org.apache.flink.api.common.state.ListState;
import org.apache.hadoop.hbase.client.Connection;

import java.util.List;
import java.util.Map;

//远期近期分段查询（行为）远用clickhouse
public class NearFarSegmentQueryRouter {
    Connection hbaseConnection;
    HbaseQueryServiceImpl hbaseQueryService;

    java.sql.Connection ckConnection;
    ClickHouseQueryServiceImpl clickHouseQueryService;

    StateQueryServiceImpl stateQueryService;

    public NearFarSegmentQueryRouter(ListState<EventBean> beansState) throws Exception {
        //获取hbase连接
        hbaseConnection = ConnectionUtils.getHbaseConnection();
        hbaseQueryService = new HbaseQueryServiceImpl(hbaseConnection);
        //获取clickhouse连接
        ckConnection = ConnectionUtils.getCkConnection();
        clickHouseQueryService = new ClickHouseQueryServiceImpl(ckConnection);
        //flink内部查询
        stateQueryService = new StateQueryServiceImpl(beansState);

    }

    public boolean ruleMatch(RuleConditions rule, EventBean eventBean,ListState<EventBean> beansState) throws Exception {
        //判断当前事件是否为规则触发事件
        if(!EventParamComparator.compare(rule.getTriggerEvent(),eventBean)) return false;
        //计算画像条件是否满足
        Map<String, String> userProfileConditions = rule.getUserProfileConditions();
        Boolean profileQueryResult = false;
        if (userProfileConditions != null){
            profileQueryResult = hbaseQueryService.queryProfileConnection(eventBean.getDeviceId(), userProfileConditions);
            if (!profileQueryResult) return false;
        }
        long segmentPoint = SegmentQueryUtil.getSegmentPoint(eventBean.getTimeStamp());
        //行为次数条件是否满足
        List<EventParam> actionCountConditions = rule.getActionCountConditions();
        if (actionCountConditions != null && actionCountConditions.size() > 0) {
            for (EventParam actionCountCondition : actionCountConditions) {

                //截至点位于分割点左方，完全用clickhouse查询
                if (actionCountCondition.getTimeRangeEnd() < segmentPoint){
                    long l = clickHouseQueryService.queryEventCountCondition(eventBean.getDeviceId(), actionCountCondition,actionCountCondition.getTimeRangeStart(),actionCountCondition.getTimeRangeEnd());
                    if (l < actionCountCondition.getCountThresHold()) return false;
                } else if (actionCountCondition.getTimeRangeStart() >= segmentPoint){//截至点位于分割点右方，用flinkSate查询
                    long count = stateQueryService.stateQueryEventCount(beansState, actionCountCondition, actionCountCondition.getTimeRangeStart(), actionCountCondition.getTimeRangeEnd());
                    if (count < actionCountCondition.getCountThresHold()) return false;
                } else { //跨界
                    //先查flinkstate
                    long countState = stateQueryService.stateQueryEventCount(beansState, actionCountCondition, segmentPoint, actionCountCondition.getTimeRangeEnd());
                    if (countState < actionCountCondition.getCountThresHold()){
                        //再查clickhouse
                        long countClickhouse = clickHouseQueryService.queryEventCountCondition(eventBean.getDeviceId(), actionCountCondition, actionCountCondition.getTimeRangeStart(), segmentPoint);
                        if (countClickhouse + countState < actionCountCondition.getCountThresHold()) return false;
                    }
                }

            }
        }
        //行为序列次数是否满足
        List<EventSequenceParam> actionSequenceCondition = rule.getActionSequenceCondition();
        if (actionSequenceCondition != null && actionSequenceCondition.size() >0){
            for (EventSequenceParam eventSequenceParam : actionSequenceCondition) {
                //截至点位于分割线左方，完全用clickhouse查询
                if (eventSequenceParam.getTimeRangeEnd() < segmentPoint){
                    int maxStep = clickHouseQueryService.queryEventSequenceCondition(eventBean.getDeviceId(), eventSequenceParam,eventSequenceParam.getTimeRangeStart(),eventSequenceParam.getTimeRangeEnd());
                    if (maxStep < eventSequenceParam.getEventSequence().size()) return false;
                }
                //截至点位于分割线右方，完全用flinkState查询
                else if (eventSequenceParam.getTimeRangeStart() > segmentPoint){
                    int i = stateQueryService.queryEventSequence(eventSequenceParam.getEventSequence(),eventSequenceParam.getTimeRangeStart(),eventSequenceParam.getTimeRangeEnd());
                    if (i < eventSequenceParam.getEventSequence().size()) return false;
                }
                //跨界
                else{
                    //先查clickhouse,确定能查到几个
                    int step1 = clickHouseQueryService.queryEventSequenceCondition(eventBean.getDeviceId(), eventSequenceParam, eventSequenceParam.getTimeRangeStart(), segmentPoint);
                    //clickhouse确定不了的规则，准备查flink
                    if (step1 < eventSequenceParam.getEventSequence().size()) {
                        List<EventParam> trimedEventParam = eventSequenceParam.getEventSequence().subList(step1, eventSequenceParam.getEventSequence().size());

                        int step2 = stateQueryService.queryEventSequence(trimedEventParam, segmentPoint, eventSequenceParam.getTimeRangeEnd());
                        if (step1 + step2 < eventSequenceParam.getEventSequence().size()) return false;
                    }
                }

            }
        }
        return true;
    }

}
